/*******************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.        */
/*                                                                 */
/*  NAME      = frtask_do.c                                        */
/*  NOTE      = 任务处理                                           */
/*  DATE      = 2014/10/23                                         */
/*******************************************************************/

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/prctl.h>

#include <exception>
using namespace std;

#include "frcomm.h"
#include "frlog.h"
#include "frmem.h"
#include "frstr.h"
#include "frsutl.h"

#include "conntrack.h"
#include "module.h"
#include "frame.h"
#include "frame_params.h"
#include "frct_comm.h"
#include "frct_mng.h"
#include "frct_expect.h"
#include "frtask_mng.h"
#include "frmodule_mng.h"
#include "frproto_detect.h"
#include "frreport_mng.h"
#include "frstat_system.h"
#include "frstat_thread.h"
#include "frstat_server.h"
#include "frstat_conn.h"
#include "frlua_config.h"

#include "frreport_comm.h"

/* response format最大重复次数 */
#define APM_L7_RESPONSE_MAX_TIME    100

/*******************************************************************************
*  FUNC     :  关联链路的处理
*  ARGS     :  p_st_report_data     (Frreport_data *,数据保存到该结构体)
*           :  p_st_ct              (Conntrack *)
*           :  p_st_data_in         (Analyz_data_in *, 协议判断用数据)
*  RTN      :
*  NOTE     :
*******************************************************************************/
//FR_STATIC void __frtask_do_relative(Conntrack *p_st_ct
//                                  , Analyz_data_in *p_st_data_in)
//{
//    L7_analyzer *p_st_l7_analyzer=NULL;
//    L7_module *p_st_l7_module=NULL;
//    u_short us_proto =0, us_port_srv=0;
//    u_int ui_ip_cli=0, ui_ip_srv=0;
//    Frreport_data st_report_data;
//    int i_ret_fun=0;
//
//    if(p_st_ct==NULL || p_st_data_in==NULL)
//        return;
//
//    p_st_l7_analyzer = frct_get_L7analyzer(p_st_ct);
//    if(p_st_l7_analyzer)
//        p_st_l7_module = p_st_l7_analyzer->p_st_l7_module;
//    if( p_st_l7_module == NULL || p_st_l7_module->l7_redirct_info == NULL)
//        return;
//    i_ret_fun = p_st_l7_module->l7_redirct_info(p_st_ct, &us_proto, &ui_ip_cli, &ui_ip_srv, &us_port_srv);
//    if(i_ret_fun == FR_FAIL)
//        return;
//
//    //加入到关联连接里
//    FRPRINT_L4(us_proto, ui_ip_cli, 0, ui_ip_srv, us_port_srv);
//    //i_ret_fun = frct_expect_add(us_proto, ui_ip_cli, ui_ip_srv, us_port_srv , p_st_ct);
//    //frdbg("frct_expect_add ret=%d\n", i_ret_fun);
//    //frct_expect_output(NULL);
//
//    memset(&st_report_data, 0x00, sizeof(Frreport_data));
//    st_report_data.em_data_type = FRREPORT_TYPE_CONN_EXPECT;
//    i_ret_fun = frreport_mng_add_conn_expect(&st_report_data, us_proto, ui_ip_cli, ui_ip_srv, us_port_srv
//                                           , p_st_ct->dst_ip, p_st_ct->dst_port, p_st_data_in->st_timestamp);
//    if(i_ret_fun == FR_FAIL)
//        return;
//
//    frreport_mng_save_report_data(p_st_ct, &st_report_data);
//
//    return;
//}

/*******************************************************************************
*  FUNC     :  释放Analyz_data_in数据
*  ARGS     :  p_st_packet       (Analyz_data_in *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __franalyz_data_free(Apm_packet *p_st_packet)
{
    if(p_st_packet == NULL)
        return;

    if(p_st_packet->p_payload)
    {
        frfree(p_st_packet->p_payload);
        p_st_packet->p_payload = NULL;
    }
    frfree(p_st_packet);

    return;
}

/*******************************************************************************
*  FUNC     :  Frreport_data节点内存释放
*  ARGS     :  p_st_node  (Frlist_node *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __list_report_node_free(Frlist_node *p_st_node)
{
    Frreport_data *p_st_report_data=NULL;
    if(p_st_node==NULL || p_st_node->data == NULL )
        return;

    p_st_report_data =  (Frreport_data*)FRLIST_GET_NODE_DATA(p_st_node);
    if(p_st_report_data)
    {
        frreport_data_free(&p_st_report_data);
    }

    return;
}

/*******************************************************************************
*  FUNC     :  统计连接信息
*  ARGS     :  p_st_ct          (Conntrack *)
*           :  p_st_cap_data    (Apm_packet *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
//FR_STATIC void __frreport_conn_info(Conntrack *p_st_ct, Apm_packet *p_st_cap_data)
//{
//    L7_analyzer *p_st_l7_analyzer = NULL;
//
//    if(p_st_ct==NULL || p_st_ct->flg_direct_known == FR_DIRECT_UNKNOWN || p_st_cap_data == NULL )
//        return;
//
//    //上报连接关闭信息(主要是tcp的fin)
//    if(p_st_ct->flg_report_conn_info == FR_YES && p_st_cap_data->em_analyz_cmd == FRANALYZ_CMD_FIN)
//    {
//        frstat_conns_del_conn_ct(p_st_ct, FR_NO);
//        return;
//    }
//
//    //上报连接开始信息
//    if(p_st_ct->flg_report_conn_info == FR_YES)
//        return;
//
//    p_st_l7_analyzer = frct_get_L7analyzer(p_st_ct);
//    if(p_st_l7_analyzer == NULL || p_st_l7_analyzer->p_st_l7_module == NULL )
//        return;
//    //上报连接开始信息
//    if(p_st_l7_analyzer->p_st_l7_module->i_l7_proto != APM_L7_UNKNOWN )
//    {
//        frstat_conns_add_conn_ct(p_st_ct, FR_YES, p_st_ct->src_ip, p_st_ct->dst_ip, p_st_ct->dst_port, 1);
//        return;
//    }
//
//    return;
//}

/*******************************************************************************
*  FUNC     :  协议判断与协议数据分析
*  ARGS     :  p_st_ct (Conntrack *)
*           :  p_st_packet  (Analyz_data_in *, 协议判断用数据)
*  RTN      :  
*  NOTE     :  操作p_st_ct上数据时,需要使用p_st_ct->lock锁
*******************************************************************************/
FR_STATIC void __frtask_analyzer(Conntrack *p_st_ct, Apm_packet *p_st_packet)
{
    int i_ret_fun=0, i_ret_analyze=0;
    int i_loop=0;
    Frhash_header st_hash_report;       /* 保存Frreport_data */
    Frlist_header *p_list_report=NULL;
    Frlist_node   *p_st_node_report=NULL;
    Frreport_data *p_st_report_data=NULL;
    L7_analyzer *p_st_l7_analyzer=NULL;

    if(frhash2_create(&st_hash_report, 2) != FR_SUCC)
    {
        FRLOG_ERROR_DETAIL("frhash2_create err");
        return;
    }

    pthread_spin_lock(&p_st_ct->lock);
    //有协议强制指定或用户指定的协议的更新,故已经识别过的协议再判断
    try
    {
        i_ret_fun = frproto_l7_detect(p_st_ct, p_st_packet);
    }
    catch (exception& e)
    {
        i_ret_fun = FR_FAIL;
    }
    //frwarn("frproto_l7_detect = %d\n", i_ret_fun);
    pthread_spin_unlock(&p_st_ct->lock);
    if(i_ret_fun == FR_FAIL)
        goto END;

    //如果是有peek数据进行的detect,则从第1个数据包开始,重新分析
    if( p_st_ct->flg_cap_peek_end == FR_NO )
    {
        p_st_ct->flg_cap_peek_end = FR_YES;
        goto END;
    }

    //上报连接信息
    //pthread_spin_lock(&p_st_ct->lock);
    //__frreport_conn_info(p_st_ct, p_st_packet);
    //pthread_spin_unlock(&p_st_ct->lock);

    //协议分析
    try
    {
        i_ret_analyze = frproto_l7_analyze(p_st_ct, p_st_packet, &st_hash_report);
    } catch (exception& e)
    {
        i_ret_analyze = APM_L7_ANALYZE_TODO;
    }
    //frwarn("data len=%d\tfrproto_l7_analyze ret=0x%x\n", p_st_packet->ui_payload_len, i_ret_analyze);

    if(i_ret_analyze == APM_L7_ANALYZE_TODO)
        goto REPORT;

    //增加统计信息
    p_st_l7_analyzer = frct_get_L7analyzer(p_st_ct);
    if(p_st_l7_analyzer && p_st_l7_analyzer->p_st_l7_module)
    {
        //增加login logout的统计信息
        if(APM_L7_ANALYZE_IS_LOGIN(i_ret_analyze))
            frstat_system_module_session_add(p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 1, 0);
        else if(APM_L7_ANALYZE_IS_LOGOUT(i_ret_analyze))
            frstat_system_module_session_add(p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 1);

        //增加服务节点的统计信息
        if(APM_L7_ANALYZE_IS_REQ(i_ret_analyze))
            frstat_server_add(p_st_ct->dst_ip, p_st_ct->dst_port, p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 1, 0);
        else if(APM_L7_ANALYZE_IS_RESP(i_ret_analyze))
            frstat_server_add(p_st_ct->dst_ip, p_st_ct->dst_port, p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 0, 1);
    }

    //增加线程统计信息
    if( p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE)
    {
        if(i_ret_analyze == APM_L7_ANALYZE_FAIL)
            frstat_thread_add(pthread_self(), 0, 1, 0, 0);
        else if(APM_L7_ANALYZE_IS_REQ(i_ret_analyze) )
            frstat_thread_add(pthread_self(), 0, 0, 1, 0);
        else if(APM_L7_ANALYZE_IS_RESP(i_ret_analyze) )
            frstat_thread_add(pthread_self(), 0, 0, 0, 1);
    }

    //(协议解析失败)/(协议解析成功，但只有1个分析协议)，程序结束
    if(i_ret_analyze == APM_L7_ANALYZE_FAIL)
        goto REPORT;

    /* 返回协议重新判断, 当前处理终止, 释放当前协议 */
    if(i_ret_analyze == APM_L7_ANALYZE_REDETECT)
    {
        pthread_spin_lock(&p_st_ct->lock);
        frct_l7analyzer_free(&p_st_ct->st_l7_analyzer);
        frct_l7analyzer_set(&p_st_ct->st_l7_analyzer, NULL);
        pthread_spin_unlock(&p_st_ct->lock);
        frct_l7proto_change_status(p_st_ct, APM_MODULE_STATUS_UNKNOWN, 0, APM_L7PROTO_STATUS_UNKNOWN);
        goto END;
    }

REPORT:
    //处理上报数据
    FRHASH_LOOP(i_loop, &st_hash_report, p_list_report)
    {
        if(p_list_report == NULL)
            continue;
        p_st_node_report = FRLIST_FIRST_NODE(p_list_report);
        while(p_st_node_report)
        {
            p_st_report_data =  (Frreport_data*)FRLIST_GET_NODE_DATA(p_st_node_report);
            if(p_st_report_data)
            {
                frreport_mng_save_report_data(p_st_ct, &p_st_report_data);
                p_st_node_report->data = NULL; //p_st_report_data数据不管成功失败,都已经释放
            }
            p_st_node_report = FRLIST_NEXT_NODE(p_st_node_report);
        }
    }

END:
    frhash2_free(&st_hash_report, __list_report_node_free);

    return ;
}

/*******************************************************************************
*  FUNC     :  reset数据包时,上报reset事件
*  ARGS     :  p_st_ct (Conntrack *)
*           :  p_st_packet  (Analyz_data_in *, 协议判断用数据)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __frtask_report_update(Conntrack *p_st_ct, Apm_packet *p_st_packet)
{
    Frreport_data *p_st_report_data=NULL;

    if(p_st_packet->i_direction != FR_DIRECT_CLIENT2SRV
        && p_st_packet->i_direction != FR_DIRECT_SRV2CLIENT)
        return;

    pthread_spin_lock(&p_st_ct->lock);
    if(p_st_packet->i_direction == FR_DIRECT_CLIENT2SRV)
        p_st_report_data = frreport_update_create(p_st_ct, APM_REPORT_UPDATE_CLIENT_ABORT, "client abort", &p_st_packet->st_timestamp);
    else
        p_st_report_data = frreport_update_create(p_st_ct, APM_REPORT_UPDATE_SERVER_ABORT, "server abort", &p_st_packet->st_timestamp);
    pthread_spin_unlock(&p_st_ct->lock);

    if(p_st_report_data)
        frreport_mng_save_report_data2(&p_st_report_data);

    return;
}

/*******************************************************************************
*  FUNC     :  协议判断与协议数据分析
*  ARGS     :  p_st_ct (Conntrack *)
*           :  p_st_packet  (Analyz_data_in *, 协议判断用数据)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __frtask_do_packet(unsigned int ui_is_peek, Conntrack *p_st_ct, Apm_packet *p_st_packet)
{
    if(p_st_ct == NULL || p_st_packet == NULL)
        return ;

    if(p_st_packet->em_analyz_cmd == FRANALYZ_CMD_RESET
       && p_st_ct->flg_direct_known == FR_DIRECT_KNOWN
       && ui_is_peek == FR_NO)
        __frtask_report_update(p_st_ct, p_st_packet);
    else
        __frtask_analyzer(p_st_ct, p_st_packet);

    return;
}

/*******************************************************************************
*  FUNC     :  调整数据包(Analyz_data_in)方向
*  ARGS     :  p_st_ct          (Conntrack *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __frcap_chg_direct(Conntrack *p_st_ct)
{
    Frlist_node *p_st_node_cap=NULL;
    Apm_packet *p_st_packet=NULL;
    int i_direction=0;

    if(p_st_ct == NULL)
        return;

    //更改 数据包缓存队列里的数据包方向
    if( frlist_size(&p_st_ct->lst_cap_queue) > 0 )
    {
        p_st_node_cap = FRLIST_FIRST_NODE(&p_st_ct->lst_cap_queue);
        while(p_st_node_cap)
        {
            p_st_packet = (Apm_packet *)FRLIST_GET_NODE_DATA(p_st_node_cap);
            if(p_st_packet == NULL)
            {
                p_st_node_cap = FRLIST_NEXT_NODE(p_st_node_cap);
                continue;
            }

            i_direction = frcap_get_cs_direction(p_st_ct, p_st_packet->src_ip, p_st_packet->src_port);
            if( i_direction== FR_CLIENT_IP_PORT)
                p_st_packet->i_direction = FR_DIRECT_CLIENT2SRV;
            else if( i_direction== FR_SRV_IP_PORT)
                p_st_packet->i_direction = FR_DIRECT_SRV2CLIENT;

            p_st_node_cap = FRLIST_NEXT_NODE(p_st_node_cap);
        }
    }

    return ;
}

/*******************************************************************************
*  FUNC     :  ct上获取数据包
*  ARGS     :  p_st_ct          (Conntrack *)
*           :  p_ui_is_peek     (标识当前数据包是否为peek; FR_YES:peek数据包)
*  RTN      :  数据包(Apm_packet *)
*  NOTE     :
*******************************************************************************/
Apm_packet *__frtask_get_packet(Conntrack *p_st_ct, unsigned int *p_ui_is_peek)
{
    Frlist_node *p_st_node_cap=NULL;
    Apm_packet *p_st_packet=NULL;

    if(p_st_ct == NULL)
        return NULL;

    //数据peek
    if(p_st_ct->flg_cap_peek_end == FR_NO)
    {
        p_st_node_cap = frlist_node_idx(&p_st_ct->lst_cap_queue, p_st_ct->ui_cap_peek_num);
        if(p_st_node_cap)
        {
            p_st_ct->ui_cap_peek_num += 1;
            if(p_st_ct->ui_cap_peek_num >= st_apm_params.ui_l7_detect_max_times)
                p_st_ct->flg_cap_peek_end = FR_YES;

            p_st_packet = (Apm_packet *)FRLIST_GET_NODE_DATA(p_st_node_cap);
            if(p_st_packet)
            {
                //数据直接获取
                if(p_st_packet->em_analyz_cmd == FRANALYZ_CMD_FIN || p_st_packet->em_analyz_cmd == FRANALYZ_CMD_RESET)
                {
                    p_st_ct->flg_cap_peek_end = FR_YES;
                    goto GET_PACKET;
                }

                if(p_ui_is_peek)
                    *p_ui_is_peek = FR_YES;
                return p_st_packet;
            }
        }
        return NULL;
    }

    //数据直接获取
GET_PACKET:
    if(p_ui_is_peek)
        *p_ui_is_peek = FR_NO;
    p_st_node_cap = frlist_pop(&p_st_ct->lst_cap_queue);
    if(p_st_node_cap)
    {
        p_st_packet = (Apm_packet *)FRLIST_GET_NODE_DATA(p_st_node_cap);
        frfree(p_st_node_cap);
        if(p_st_packet)
            return p_st_packet;
    }

    return NULL;
}

/*******************************************************************************
*  FUNC     :  执行工作任务
*  ARGS     :  
*  RTN      :  
*  NOTE     :  
*******************************************************************************/
void *frtask_do(void *arg)
{
    Conntrack *p_st_ct = NULL;
    Apm_packet *p_st_packet=NULL;
    unsigned int ui_is_peek=FR_NO;
    char arr_thread_name[17];

    //FRLOG_DEBUG("frtask_do(%l) run...", pthread_self());
    FRLOG_DEBUG("frtask_do run...");
    frdbg("frtask_do(%lu) run...\n", pthread_self());
    prctl(PR_SET_NAME,"task_do");
    while(g_i_flag_run)
    {
        //frdbg("frtask_get start\n");
        p_st_ct = frtask_get();
        //frdbg("frtask_get end(%p)\n", p_st_ct);
        if(p_st_ct == NULL)
        {
            usleep(10000);
            continue;
        }
        
        //获取解析数据并处理
        do
        {
            pthread_spin_lock(&p_st_ct->lock);
            if(p_st_ct->flg_task_out == 1)
            {
                p_st_ct->flg_task_out = 0;
                frct_put(p_st_ct);
                p_st_ct->flg_inTask = FR_NO;
                frtask_add(p_st_ct);
                pthread_spin_unlock(&p_st_ct->lock);
                break;
            }

            ui_is_peek=FR_NO;
            p_st_packet = __frtask_get_packet(p_st_ct, &ui_is_peek);
            if(p_st_packet == NULL)
            {
                frct_put(p_st_ct);
                p_st_ct->flg_inTask = FR_NO;
            }
            pthread_spin_unlock(&p_st_ct->lock);
            if(p_st_packet == NULL)
                break;
            //FRPRINT_L4(p_st_ct->ip_proto
            //         , p_st_packet->src_ip, p_st_packet->src_port, p_st_packet->dst_ip, p_st_packet->dst_port
            //         , p_st_packet->ui_payload_len, p_st_packet->em_analyz_cmd);
            //FRPRINT_BIN("frtask_do", p_st_packet->p_payload, p_st_packet->ui_payload_len);printf("\n");

            #ifdef FRDBG_CAP_DO
            {
                char arr_buff[128], arr_buff_time[30];

                arr_buff[0] = '\0';
                frl4_info(p_st_ct->ip_proto
                        , p_st_packet->src_ip, p_st_packet->src_port, p_st_packet->dst_ip, p_st_packet->dst_port
                        , p_st_packet->ui_payload_len, p_st_packet->em_analyz_cmd
                        , arr_buff, sizeof(arr_buff));
                snprintf(arr_buff+strlen(arr_buff), sizeof(arr_buff)-strlen(arr_buff), " ct_direct=%d", p_st_ct->flg_direct_known);
                arr_buff_time[0] = '\0';
                frtime2str(&p_st_packet->st_timestamp, arr_buff_time, sizeof(arr_buff_time));
                FRLOG_DEBUG("cap_info:%s time:%s", arr_buff, arr_buff_time);
            }
            #endif

            //数据解析
            p_st_ct->ui_done_data_cnt +=1;
            if(p_st_ct->st_l7_analyzer.p_st_l7_module)
            {
                snprintf(arr_thread_name, sizeof(arr_thread_name), "task_%d_%s"
                        , p_st_ct->st_l7_analyzer.p_st_l7_module->i_l7_proto
                        , p_st_ct->st_l7_analyzer.p_st_l7_module->p_version);
                prctl(PR_SET_NAME, arr_thread_name);
            }
            else
                prctl(PR_SET_NAME,"task_do_in");

            __frtask_do_packet(ui_is_peek, p_st_ct, p_st_packet);
            prctl(PR_SET_NAME,"task_do_out");

            //增加线程统计信息
            frstat_thread_add(pthread_self(), 1, 0, 0, 0);

            //如果更改ct连接方向,调整缓存数据包中的flg_direct_known
            pthread_spin_lock(&p_st_ct->lock);
            if(p_st_ct->flg_chg_direct == FR_YES || p_st_ct->flg_chg_stat_ip == 2)
            {
                //{
                //    char arr_buff[128];
                //    frl4_print2(arr_buff, sizeof(arr_buff), p_st_ct->ip_proto, p_st_ct->src_ip, p_st_ct->src_port, p_st_ct->dst_ip, p_st_ct->stat_dst_port);
                //    FRLOG_ERROR_DETAIL("%s ",arr_buff);
                //}
                if(p_st_ct->stat_dst_port != 0)
                    frstat_conns_add_conn_ct(p_st_ct, FR_YES, p_st_ct->src_ip, p_st_ct->dst_ip, p_st_ct->stat_dst_port, 1, 0, 0);
                else
                    frstat_conns_add_conn_ct(p_st_ct, FR_YES, p_st_ct->src_ip, p_st_ct->dst_ip, p_st_ct->dst_port, 1, 0, 0);
                if(p_st_ct->flg_chg_stat_ip == 2)
                    p_st_ct->flg_chg_stat_ip = 1;
            }

            if(p_st_ct->flg_chg_direct == FR_YES)
            {
                p_st_ct->flg_chg_direct = FR_NO;
                __frcap_chg_direct(p_st_ct);
            }
            pthread_spin_unlock(&p_st_ct->lock);

            //peek数据包不释放
            if(ui_is_peek == FR_YES)
                continue;

            //释放数据包
            if(p_st_packet)
            {
                __franalyz_data_free(p_st_packet);
                __sync_fetch_and_sub(&g_ul_pcap_queue_used, 1);
            }

        } while(g_i_flag_run);

        if(g_i_flag_run == 0 && p_st_ct->flg_inTask == FR_YES)
        {
            frct_put(p_st_ct);
            p_st_ct->flg_inTask = FR_NO;
        }
    }
    
    pthread_exit(0);
    return (void*)NULL;
}

